#include <list>
#include <staff/common/DataObject.h>
#include <staff/common/DataObjectHelper.h>
#include <staff/common/Operation.h>
#include <rise/socket/ClientSocket.h>
#include <rise/common/StreamBuffer.h>
#include <rise/threading/CriticalSection.h>
#include <rise/threading/Thread.h>
#include <rise/common/ExceptionTemplate.h>
#include <rise/common/exmacros.h>
#include "Message.h"
#include "Client.h"

namespace smq
{

  struct Client::ClientImpl
  {
    typedef std::list<staff::CDataObject> TMessageList;

    rise::sockets::CClientSocket tSocket;
    rise::CStreamBuffer tSendBuffer;
    rise::CStreamBuffer tRecvBuffer;
    bool m_bShuttingDown;
    rise::threading::CCriticalSection m_tSendLock;
    rise::threading::CCriticalSection m_tRecvLock;
    TMessageList m_lsSavedMessages;
    
    ClientImpl():
      m_bShuttingDown(false)
    {
    }

    bool GetSavedMessage(staff::CDataObject& rMsg, bool bIsMessage)
    {
      // looking for stored messages
      for (TMessageList::iterator itMessage = m_lsSavedMessages.begin();
            itMessage != m_lsSavedMessages.end(); ++itMessage)
      {
        const std::string sMsgName = itMessage->GetLocalName();
        if (bIsMessage == (sMsgName == "Message"))
        {
          rMsg = *itMessage;
          m_lsSavedMessages.erase(itMessage);
          return true;
        }
      }
      return false;
    }
  };


  Client::Client()
  {
    m_pImpl = new ClientImpl;
  }

  Client::~Client()
  {
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  void Client::Init(const std::string& sServerHost /*= "127.0.0.1"*/, unsigned short ushServerPort /*= 8768*/)
  {
    m_pImpl->m_bShuttingDown = false;
    m_pImpl->tSocket.Create();
    m_pImpl->tSocket.Connect(sServerHost, ushServerPort);
    m_pImpl->tSendBuffer.Reserve(4096);
    m_pImpl->tRecvBuffer.Reserve(40960);
    m_pImpl->m_lsSavedMessages.clear();
  }
  
  void Client::Shutdown()
  {
    if (!m_pImpl->m_bShuttingDown)
    {
      m_pImpl->m_bShuttingDown = true;
      m_pImpl->tSocket.Shutdown(rise::sockets::CSocket::ES_BOTH);
    }
  }

  bool Client::IsShuttingDown() const
  {
    return m_pImpl->m_bShuttingDown;
  }

  void Client::Close()
  {
    m_pImpl->tSocket.Close();
  }

  void Client::Post(const staff::CDataObject& rMsg)
  {
    rise::threading::CScopedCriticalSection tLock(m_pImpl->m_tSendLock);

    m_pImpl->tSendBuffer.Reset();
    m_pImpl->tSendBuffer << rMsg;

    // sending request
    rise::ulong ulSent = 0;
    rise::ulong ulBuffPos = 0;
    rise::ulong ulDataSize = m_pImpl->tSendBuffer.GetSize();

#ifdef _DEBUG
      rise::LogDebug3() << "Sending: " << ulDataSize << " bytes:\n" <<
        "\033[34m" << std::string(reinterpret_cast<const char*>(m_pImpl->tSendBuffer.GetData()), ulDataSize) << "\033[0m";
#endif
    while (ulDataSize > 0)
    {
      if (m_pImpl->m_bShuttingDown)
      {
        rise::LogDebug1() << "Client was closed. breaking posting...";
        return;
      }
      m_pImpl->tSocket.Send(m_pImpl->tSendBuffer.GetData() + ulBuffPos, ulDataSize, &ulSent);
      ulBuffPos += ulSent;
      ulDataSize -= ulSent;
    }
  }

  void Client::Receive(staff::CDataObject& rMsg, bool bIsMessage)
  {
#ifdef _DEBUG
    rise::LogDebug2() << "receiving " << (bIsMessage ? "message" : "response") << "...";
#endif
    if (m_pImpl->GetSavedMessage(rMsg, bIsMessage))
    {
      return;
    }

    // receiving message
    for(;;)
    {
      { // critical section scope
        rise::threading::CScopedCriticalSection tLock(m_pImpl->m_tRecvLock);
        if (m_pImpl->GetSavedMessage(rMsg, bIsMessage))
        {
          return;
        }

        staff::CDataObject tdoMsg;

        // receiving response
        bool bRcvd = true;
        rise::ulong ulCurrentSize = 0;
        rise::ulong ulReceived = m_pImpl->tRecvBuffer.GetSize();
        m_pImpl->tRecvBuffer.Normalize();
        while (bRcvd)
        {
          if(memchr(m_pImpl->tRecvBuffer.GetData() + ulCurrentSize, '\0', ulReceived) != NULL)
          {
            ulCurrentSize += ulReceived;
            break;
          }

          ulCurrentSize += ulReceived;

          bRcvd = m_pImpl->tSocket.Recv(m_pImpl->tRecvBuffer, true, &ulReceived);

          if (m_pImpl->m_bShuttingDown)
          {
            rise::LogDebug1() << "Client was closed. breaking receiving...";
            return;
          }

          if (ulReceived == 0)
          {
            m_pImpl->tSocket.Close();
            rise::LogDebug() << "Connection Closed";
            RISE_THROWS(rise::CFileCloseException, "Connection closed");
          }
        } // while (bRcvd)


#ifdef _DEBUG
        rise::LogDebug3() << "received: " << ulCurrentSize << " bytes:\n" <<
          "\033[35m" << std::string(reinterpret_cast<const char*>(m_pImpl->tRecvBuffer.GetData()), ulCurrentSize) << "\033[0m";
#endif

        m_pImpl->tRecvBuffer >> tdoMsg;

        {
          const std::string sMsgName = tdoMsg.GetLocalName();
          if (bIsMessage == (sMsgName == "Message"))
          {
            rMsg = tdoMsg;
            break;
          }
          m_pImpl->m_lsSavedMessages.push_back(tdoMsg);
        }
      }
      // switching to another thread
      rise::threading::CThread::Sleep(0);
    }
  }

}
